package cn.linjianhui.flink.sample;

import cn.linjianhui.flink.sample.model.AreaAccumulator;
import cn.linjianhui.flink.sample.model.OmsOrder;
import cn.linjianhui.flink.sample.util.ParamUtils;
import cn.linjianhui.flink.sample.util.Utils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousProcessingTimeTrigger;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

import java.math.BigDecimal;
import java.time.LocalDate;
import java.time.ZoneOffset;
import java.util.Properties;

public class AreaPayAmountJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = Utils.getStreamExecutionEnvironment();

        Properties props = ParamUtils.getKafkaConsumerProperty(args, "area-pay", "earliest");

        DataStream<String> sourceStream = env.addSource(new FlinkKafkaConsumer011<>("OMS-Order", new SimpleStringSchema(), props));

        DataStream<OmsOrder> payOrderStream = sourceStream
                .map((MapFunction<String, OmsOrder>) s -> JSON.parseObject(s, OmsOrder.class))
                .filter((FilterFunction<OmsOrder>) omsOrder -> {
                    long currentDayTimestamp = LocalDate.now().atStartOfDay(ZoneOffset.ofHours(8)).toInstant().toEpochMilli();
                    return omsOrder.getORDERSTATUS() == 3 && omsOrder.getLASTMODIFYTIME() >= currentDayTimestamp;
                });

        DataStream<AreaAccumulator> areaAccumulatorStream = payOrderStream
                .keyBy("PROVINCECODE")
                .window(TumblingProcessingTimeWindows.of(Time.days(1), Time.hours(-8)))
                .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(5)))
                .aggregate(new AggregateFunction<OmsOrder, AreaAccumulator, AreaAccumulator>() {
                    @Override
                    public AreaAccumulator createAccumulator() {
                        return new AreaAccumulator();
                    }

                    @Override
                    public AreaAccumulator add(OmsOrder order, AreaAccumulator accumulator) {
                        if(!accumulator.containsOrder(order.getORDERCODE())) {
                            accumulator.setProvince(order.getPROVINCE());
                            accumulator.setProvinceCode(order.getPROVINCECODE());
                            accumulator.addPayAmount(new BigDecimal(order.getNETAMT()));
                            accumulator.addOrderId(order.getORDERCODE());
                        }
                        return accumulator;
                    }

                    @Override
                    public AreaAccumulator getResult(AreaAccumulator result) {
                        return result;
                    }

                    @Override
                    public AreaAccumulator merge(AreaAccumulator acc1, AreaAccumulator acc2) {
                        acc1.addPayAmount(acc2.getPayAmount());
                        return acc1;
                    }
                });

        FlinkJedisPoolConfig jedisPoolConfig = ParamUtils.getJedisPoolConfig(args);
        areaAccumulatorStream.addSink(new RedisSink<>(jedisPoolConfig, new RedisMapper<AreaAccumulator>() {
            @Override
            public RedisCommandDescription getCommandDescription() {
                return new RedisCommandDescription(RedisCommand.HSET, "flink-sample-area_pay");
            }

            @Override
            public String getKeyFromData(AreaAccumulator accumulator) {
                return accumulator.getProvinceCode();
            }

            @Override
            public String getValueFromData(AreaAccumulator accumulator) {
                JSONObject json = new JSONObject();
                json.put("province", accumulator.getProvince());
                json.put("provinceCode", accumulator.getProvinceCode());
                json.put("payAmount", accumulator.getPayAmount().setScale(4, BigDecimal.ROUND_HALF_EVEN));
                return json.toJSONString();
            }
        }));

        areaAccumulatorStream.map((MapFunction<AreaAccumulator, String>) JSON::toJSONString).print();

        env.execute("PayAmount By Area Job");
    }
}
